package com.bw.gmall.realtime.app.zyjdwd;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class ZwTest1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> date = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("topic_log", "gmall-realtime-2020" ));
        OutputTag<String> data_error = new OutputTag<String>("data_error"){};
//        SingleOutputStreamOperator<String> datefilter = date.filter(new FilterFunction<String>() {
//            @Override
//            public boolean filter(String s) throws Exception {
//                boolean valid = JSONObject.isValid(s);
//                if (!valid) {
//                    return false;
//                }
//                if (JSONObject.parseObject(s).getJSONObject("common").getString("mid") == null || JSONObject.parseObject(s).getString("common") == null) {
//                    return false;
//                }
//                return true;
//            }
//        });
//        datefilter.print("page>>>>>>>>>>");
        SingleOutputStreamOperator<JSONObject> dateflit2 = date.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                boolean valid = JSONObject.isValid(value);
                if (!valid) {
                    ctx.output(data_error, value);
                } else {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    if (jsonObject.getJSONObject("common").getString("mid") == null || jsonObject.getString("common") == null) {
                        ctx.output(data_error, value);
                    } else {
                        out.collect(jsonObject);
                    }
                }
            }
        });
//        dateflit2.print("dateflit2>>>>>>>>>>");
   //        dateflit2.getSideOutput(data_error).print("error>>>>>>>>>>");
   //    dateflit2.print("page>>>>>>>>>>");
        //新老用户状态修改
        SingleOutputStreamOperator<JSONObject> is_new_xf_day = dateflit2.keyBy(a -> a.getJSONObject("common").getString("mid")).process(new ProcessFunction<JSONObject, JSONObject>() {
            ValueState<String> newdate;

            @Override
            public void open(Configuration parameters) throws Exception {
                newdate = getRuntimeContext().getState(new ValueStateDescriptor<String>("newdate", String.class));
            }


            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out)
                    throws Exception {
                Long ts = value.getLong("ts");
                String is_new = value.getJSONObject("common").getString("is_new");
                String day = DateFormatUtil.toDate( ts);
                String newdateState = newdate.value();
                if (is_new.equals("1")) {
                    if (newdateState == null) {
                        newdate.update(day);
                    } else {
                        if (!newdateState.equals(day)) {
                            is_new = "0";
                            value.getJSONObject("common").put("is_new", is_new);
                        }
                    }
                    out.collect(value);
                } else {
                    if (newdateState == null) {
                        String yesterday = DateFormatUtil.toDate(ts- 1000 * 60 * 60 * 24);
                        newdate.update(yesterday);
                    }
                    out.collect(value);
                }
            }
        });
//        is_new_xf_day.print("is_new_xf_day>>>>>>>>>>");
        OutputTag<String> pageTag = new OutputTag<String>("pageTag") {
        };
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };
        SingleOutputStreamOperator<JSONObject> main = is_new_xf_day.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String error = value.getString("err");
                if (error != null) {
                    ctx.output(errorTag, value.toJSONString());
                }
                value.remove("err");
                String start = value.getString("start");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                } else {
                    String common = value.getString("common");
                    String page = value.getString("page");
                    String ts = value.getString("ts");
                    JSONArray display = value.getJSONArray("display");
                    if (display!= null){
                        for (int i = 0; i < display.size(); i++) {
                            JSONObject displayObj = display.getJSONObject(i);
                            JSONObject jsonObj = new JSONObject();
                            jsonObj.put("display", displayObj);
                            jsonObj.put("common", common);
                            jsonObj.put("page", page);
                            jsonObj.put("ts", ts);
                            ctx.output(displayTag, jsonObj.toJSONString());
                        }
                    }
                    JSONArray actions = value.getJSONArray("action");
                    if (actions != null){
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject actionobj = actions.getJSONObject(i);
                            JSONObject jsonObj = new JSONObject();
                            jsonObj.put("action", actionobj);
                            jsonObj.put("common", common);
                            jsonObj.put("page", page);
                            jsonObj.put("ts", ts);
                            ctx.output(actionTag, jsonObj.toJSONString());
                        }
                    }
                    value.remove("displays");
                    value.remove("action");
                    ctx.output(pageTag, value.toJSONString());
                }
            }
        });
        DataStream<String> pagedata = main.getSideOutput(pageTag);
        pagedata.print("page>>>>>>>>>>");
        DataStream<String> startdata = main.getSideOutput(startTag);
        startdata.print("start>>>>>>>>>>");
        DataStream<String> displaydata = main.getSideOutput(displayTag);
        displaydata.print("display>>>>>>>>>>");
        DataStream<String> actiondata = main.getSideOutput(actionTag);
        actiondata.print("action>>>>>>>>>>");
        DataStream<String> errordata = main.getSideOutput(errorTag);
        errordata.print("error>>>>>>>>>>");
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        pagedata.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startdata.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displaydata.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actiondata.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errordata.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        env.execute();
    }
}
